package com.millstein.realtime.app.dwd.db;

import com.millstein.realtime.app.base.BaseSqlApp;
import com.millstein.realtime.common.Constants;
import com.millstein.realtime.util.SqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @Description
 * @Author tsing
 * @Date 2024-10-10 14:15
 */
public class Dwd_05_TradeOrderPreProcess extends BaseSqlApp {

    public static void main(String[] args) {
        new Dwd_05_TradeOrderPreProcess().init(
                6000,
                3,
                "Dwd_05_DwdTradeOrderPreProcess"
        );
    }

    /**
     * 具体数据处理的逻辑，由子类编写
     *
     * @param env      执行环境
     * @param tableEnv 表执行环境
     */
    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) {
        // 1.设置表执行环境
        // 设置过期时间。假设下单的支付时间为30分钟
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(30 * 60 + 5));

        // 2.从ods层读取业务数据
        readOdsDataFromKafka(tableEnv, "Dwd_05_DwdTradeOrderPreProcess");

        // 3.从数据库中读取字典表的数据
        readBaseDicFromMysql(tableEnv);

        // 4.从业务数据中筛选出order_info数据
        Table orderDetailTable = tableEnv.sqlQuery(
                "select " +
                "    `data`['id'] id, " +
                "    `data`['order_id'] order_id, " +
                "    `data`['sku_id'] sku_id, " +
                "    `data`['sku_name'] sku_name, " +
                "    `data`['create_time'] create_time, " +
                "    `data`['source_id'] source_id, " +
                "    `data`['source_type'] source_type, " +
                "    `data`['sku_num'] sku_num, " +
                "    cast(cast(`data`['sku_num'] as decimal(16, 2)) * cast(`data`['order_price'] as decimal(16, 2)) as string) split_origin_amount, " +
                "    `data`['split_total_amount'] split_total_amount, " +
                "    `data`['split_activity_amount'] split_activity_amount, " +
                "    `data`['split_coupon_amount'] split_coupon_amount, " +
                "    ts od_ts, " +
                "    pt " +
                "from maxwell_table " +
                "where `database` = 'gmall' " +
                "    and `table` = 'order_detail' " +
                "    and `type` = 'insert'"
        );
        tableEnv.createTemporaryView("order_detail", orderDetailTable);

        // 5.从业务数据中筛选出order_detail数据
        Table orderInfoTable = tableEnv.sqlQuery(
                "select " +
                "    `data`['id'] id, " +
                "    `data`['user_id'] user_id, " +
                "    `data`['province_id'] province_id, " +
                "    `data`['operate_time'] operate_time, " +
                "    `data`['order_status'] order_status, " +
                "    `type`, " +
                "    `old`, " +
                "    `ts` oi_ts " +
                "from maxwell_table " +
                "where `database` = 'gmall' " +
                "    and `table` = 'order_info' " +
                "    and (`type` = 'insert' or `type` = 'update')"
        );
        tableEnv.createTemporaryView("order_info", orderInfoTable);

        // 6.从业务数据中筛选出order_detail_activity数据
        Table activityTable = tableEnv.sqlQuery(
                "select " +
                "    `data`['order_detail_id'] order_detail_id, " +
                "    `data`['activity_id'] activity_id, " +
                "    `data`['activity_rule_id'] activity_rule_id " +
                "from maxwell_table " +
                "where `database` = 'gmall' " +
                "    and `table` = 'order_detail_activity' " +
                "    and `type` = 'insert'"
        );
        tableEnv.createTemporaryView("order_detail_activity", activityTable);

        // 7.从业务数据中筛选出order_detail_coupon数据
        Table couponTable = tableEnv.sqlQuery(
                "select " +
                "    `data`['order_detail_id'] order_detail_id, " +
                "    `data`['coupon_id'] coupon_id " +
                "from maxwell_table " +
                "where `database` = 'gmall' " +
                "    and `table` = 'order_detail_coupon' " +
                "    and `type` = 'insert'"
        );
        tableEnv.createTemporaryView("order_detail_coupon", couponTable);

        // 8.五张表数据进行join
        Table resultTable = tableEnv.sqlQuery(
                "select " +
                "    od.id, " +
                "    od.order_id, " +
                "    od.sku_id, " +
                "    od.sku_name, " +
                "    od.create_time, " +
                "    od.source_id, " +
                "    od.source_type source_type_code, " +
                "    bd.dic_name source_type_name, " +
                "    od.sku_num, " +
                "    od.split_origin_amount, " +
                "    od.split_total_amount, " +
                "    od.split_activity_amount, " +
                "    od.split_coupon_amount, " +
                "    od.od_ts, " +
                "    oi.user_id, " +
                "    oi.province_id, " +
                "    oi.operate_time, " +
                "    oi.order_status, " +
                "    oi.`type`, " +
                "    oi.`old`, " +
                "    oi.oi_ts, " +
                "    oda.activity_id, " +
                "    oda.activity_rule_id, " +
                "    odc.coupon_id, " +
                "    current_row_timestamp() row_opt_ts " +
                "from order_detail od " +
                "join order_info oi on od.order_id = oi.id " +
                "join base_dic for system_time as of od.pt as bd on od.source_type = bd.dic_code " +
                "left join order_detail_activity oda on od.id = oda.order_detail_id " +
                "left join order_detail_coupon odc on od.id = odc.order_detail_id"
        );

        // 9.创建kafka的动态表
        tableEnv.executeSql(
                "create table dwd_trade_order_pre_process ( " +
                "    id string, " +
                "    order_id string, " +
                "    sku_id string, " +
                "    sku_name string, " +
                "    create_time string, " +
                "    source_id string, " +
                "    source_type_code string, " +
                "    source_type_name string, " +
                "    sku_num string, " +
                "    split_origin_amount string, " +
                "    split_total_amount string, " +
                "    split_activity_amount string, " +
                "    split_coupon_amount string, " +
                "    od_ts bigint, " +
                "    user_id string, " +
                "    province_id string, " +
                "    operate_time string, " +
                "    order_status string, " +
                "    `type` string, " +
                "    `old` map<string, string>, " +
                "    oi_ts bigint, " +
                "    activity_id string, " +
                "    activity_rule_id string, " +
                "    coupon_id string, " +
                "    row_opt_ts timestamp_ltz(3), " +
                "    primary key (id) not enforced " +
                ")" + SqlUtil.getUpsertKafkaSinkDDL(Constants.TOPIC_DWD_TRADE_ORDER_PRE_PROCESS)
        );

        // 10.将join好的数据写入kafka中
        resultTable.executeInsert("dwd_trade_order_pre_process");
    }
}
